package com.shujia.streaming

import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo03Window {
  def main(args: Array[String]): Unit = {
    // 构建SparkContext
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo03Window")
      .master("local[2]") // 接收数据会一直占用一个线程
      .getOrCreate()

    /**
     * 创建SparkStreaming的入口
     * 需要SparkContext以及处理的间隔时间
     * 每5s处理一次---> 每5s将接收到的数据 封装成 一个RDD进行处理
     */
    val ssc: StreamingContext = new StreamingContext(spark.sparkContext, Durations.seconds(5))

    ssc.checkpoint("Spark/data/checkpoint")
    /**
     * yum install nc
     * nc -lk 8888
     */

    // 可以通过socket连接nc命令创建的服务器 模拟消息队列
    val lineDS: DStream[String] = ssc.socketTextStream("master", 8888)


    /**
     * 热门微博：每小时基于最近24小时的搜索及浏览记录统计热门微博
     * 简化成：每隔5s统计最近15s内的单词数量
     * 5s:滑动的时间
     * 15s:窗口的大小
     * 滑动时间 一般是小于 窗口的大小 ==> 滑动窗口
     * 滑动时间 如果等于 窗口的大小 ==> 滚动窗口
     * 滑动时间一般等于创建StreamingContext时指定的batch的大小
     * 需要用到窗口
     *
     */

    lineDS
      .flatMap(_.split(","))
      .map(word => (word, 1))
      //  .reduceByKey(_ + _) // 只能统计当前batch的单词数量
      //      .updateStateByKey() // 可以统计整个任务运行周期的单词数量
      .reduceByKeyAndWindow((x: Int, y: Int) => x + y // 聚合函数
        , (i: Int, j: Int) => i - j // 优化
        , Durations.seconds(15) // 窗口的大小
        , Durations.seconds(5) // 滑动的时间
      )
      .print()


    // 启动任务
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
